package com.leisurexi.gedis.client;

import com.leisurexi.gedis.client.handler.ConnectionEventHandler;
import com.leisurexi.gedis.client.handler.PacketCodecHandler;
import com.leisurexi.gedis.client.handler.ResponseHandler;
import com.leisurexi.gedis.client.protocol.RequestProtocol;
import com.leisurexi.gedis.client.protocol.ResponseProtocol;
import io.netty.bootstrap.Bootstrap;
import io.netty.buffer.PooledByteBufAllocator;
import io.netty.channel.*;
import io.netty.channel.epoll.Epoll;
import io.netty.channel.epoll.EpollEventLoopGroup;
import io.netty.channel.epoll.EpollSocketChannel;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.LengthFieldBasedFrameDecoder;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import java.util.concurrent.*;

/**
 * 客户端
 *
 * @author: leisurexi
 * @date: 2021-07-30 13:44
 */
@Component
@Slf4j
public class GedisClient {

    /**
     * 存放请求命令队列
     */
    public static final ConcurrentLinkedQueue<RequestProtocol> requestQueue = new ConcurrentLinkedQueue();

    /**
     * 存放响应队列
     */
    public static final BlockingQueue<ResponseProtocol> responseQueue = new LinkedBlockingDeque<>();

    @Value("${gedis.host}")
    private String host;

    @Value("${gedis.port}")
    private Integer port;

    private Channel channel;

    private EventLoopGroup worker;

    @PostConstruct
    private void init() {
        Executors.newSingleThreadExecutor().execute(() -> start());
    }

    @PreDestroy
    private void destroy() {
        if (this.worker != null) {
            this.worker.shutdownGracefully();
        }
    }

    private void start() {
        if (Epoll.isAvailable()) {
            worker = new EpollEventLoopGroup();
        } else {
            worker = new NioEventLoopGroup();
        }

        Bootstrap bootstrap = new Bootstrap();
        bootstrap.group(worker)
                .channel(Epoll.isAvailable() ? EpollSocketChannel.class : NioSocketChannel.class)
                .option(ChannelOption.SO_REUSEADDR, true)
                .option(ChannelOption.SO_KEEPALIVE, true)
                .option(ChannelOption.SO_SNDBUF, 1024 * 32)
                .option(ChannelOption.SO_RCVBUF, 1024 * 32)
                .option(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT)
                .handler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel ch) throws Exception {
                        ch.pipeline()
                                .addLast(new LengthFieldBasedFrameDecoder(Integer.MAX_VALUE, 3, 4))
                                .addLast(PacketCodecHandler.INSTANCE)
                                // 使用 Netty 自带的心跳处理器，每1分钟发送一次心跳
//                                .addLast(new IdleStateHandler(0, 0, 1, TimeUnit.MINUTES))
                                .addLast(ConnectionEventHandler.INSTANCE)
                                .addLast(ResponseHandler.INSTANCE);
                    }
                });

        try {
            ChannelFuture future = bootstrap.connect(host, port).sync();
            if (future.isSuccess()) {
                log.info("连接 Gedis 服务成功");
                this.channel = future.channel();
            } else {
                log.error("连接 Gedis 服务失败", future.cause());
            }
        } catch (InterruptedException e) {
            log.error(e.getMessage(), e);
        }
    }

    /**
     * 发送数据到服务端
     *
     * @param request 包数据
     */
    public ResponseProtocol send(RequestProtocol request) {
        return send(request, 0, TimeUnit.SECONDS);
    }

    public ResponseProtocol send(RequestProtocol request, long timeout, TimeUnit timeUnit) {
        if (this.channel != null) {
            requestQueue.offer(request);
            ResponseProtocol response = null;
            try {
                this.channel.writeAndFlush(request).sync();
                if (timeout > 0) {
                    response = responseQueue.poll(timeout, timeUnit);
                } else {
                    response = responseQueue.take();
                }
            } catch (InterruptedException e) {
                log.error(e.getMessage(), e);
            } finally {
                if (response != null) {
                    requestQueue.poll();
                }
            }
            return response;
        }
        throw new NullPointerException("channel is null.");
    }

}
